package design;

/**
 * CompletionService
 * TODO *任务产生线程                                                                                               结果处理线程
 * Submit Task（提交任务）                    Store Result(存储结果)                      拉取Result
 * * Completion---------------------->Executor(线程池处理)--------------------->BlockedQueue(阻塞队列)---------->获取并处理结果
 */

import design.masterWorker.SalaryCallable0;

import java.util.concurrent.*;

/** CompletionService的使用模式
 * 你可用它不断的提交任务（线程）给Executor处理
 * 处理后的记过都会自动放入BlockedQueue
 * 另外一个线程不断的从队列里获取处理结果
 * 那个任务先处理完就能先得到那个结果
 * 组后做汇总处理
 * 从而轻松万册灰姑娘MasterWorker模式相同的功能
 */
public class MasterWorkTest4 {
    // CompletionService
    // submit 用于提交任务
    // take用于获取处理结果（阻塞式）
    // poll也用于获取处理结果（非阻塞式）
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        // 创建线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(3, 10, 5,
                TimeUnit.SECONDS, new LinkedBlockingDeque<>());
        // 创建CompletionService,需要借 助Executor来实现
        CompletionService cs = new ExecutorCompletionService(executor);
        // 提交任务线程
        Thread t0 = new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 100; i++) {
                    // 创建回调线程
                    SalaryCallable0 sc1 = new SalaryCallable0(300, ThreadLocalRandom.current().nextInt(2000));
                    cs.submit(sc1);
                }
            }
        });
        t0.start();
        // 一个线程不断处理结果
        Thread t1 = new Thread(new Runnable() {
            // 阻塞-take获取已经处理结束的结果，最先处理结束的，最先得到
            Future f = null;
            @Override
            public void run() {
                while (true) {
                    try {
                        f = cs.take();
                        System.out.println(Thread.currentThread().getName() + "getResult = " + f.get());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (ExecutionException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        t1.start();

        t0.join();
        t1.join();
        executor.shutdown();
    }
}
